package Core

import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.{SparkConf, SparkContext}

object test {

    def main(args: Array[String]): Unit = {
      //设置运行环境
      val conf = new SparkConf().setAppName("Pregel API GraphX").setMaster("local")
      val sc = new SparkContext(conf)
      sc.setLogLevel("WARN")

      // 构建图
      val myVertices = sc.parallelize(Array((1L, 0), (2L, 0), (3L, 0), (4L, 0),
        (5L, 0)))
      val myEdges = sc.makeRDD(Array( Edge(1L,2L,2.5),
        Edge(2L,3L,3.6) , Edge(3L,4L,4.5),
        Edge(4L,5L,0.1),Edge(3L,5L,5.2)
      ))
      val myGraph = Graph(myVertices,myEdges)

      val sourceId: VertexId = 1L // The ultimate source
      // Initialize the graph such that all vertices except the root have distance infinity.
      val initialGraph = myGraph.mapVertices((id, _) =>
        if (id == sourceId) 0.0 else Double.PositiveInfinity)


      val sssp: Graph[Double, Double] = initialGraph.pregel(Double.PositiveInfinity)(
        (id, dist, newDist) => math.min(dist, newDist), // Vertex Program
        triplet => {  // Send Message
          if (triplet.srcAttr + triplet.attr < triplet.dstAttr) {
            Iterator((triplet.dstId, triplet.srcAttr + triplet.attr))
          } else {
            Iterator.empty
          }
        },
        (a, b) => math.min(a, b) // Merge Message
      )

      sssp.vertices.collect.foreach(println(_))
    }


  }
